package storage

import (
	"errors"
	"fmt"
	"sort"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/JMArch/rpcmon-gateway/conf"
	"gitee.com/JMArch/rpcmon-gateway/log"
	"go.uber.org/zap"
)

type CacheRequest struct {
	Send chan *Item
}

type CacheResponse struct {
	Remote  string
	Receive chan *Item
	History []*Item
}

type Node map[string]interface{}

type Item struct {
	Data   []Node
	Create time.Time
	expire time.Duration
}

func (i *Item) isExpired() bool {
	return time.Now().Sub(i.Create) >= i.expire
}

type Items struct {
	mu          sync.Mutex
	array       []*Item
	waiting     map[string]*CacheRequest
	lastHistory time.Time
	lastOne     time.Time
	uniqMap     map[string]time.Time
}

type Cache struct {
	Name          string
	log           *zap.Logger
	entries       sync.Map
	count         uint64
	wg            sync.WaitGroup
	quit          chan bool
	checkInterval time.Duration
	allHistory    time.Duration
}

type CacheGroup struct {
	rw       sync.RWMutex
	RPCGroup map[string]*Cache
	OWLGroup map[string]*Cache
}

func NewCacheGroup(config *conf.GatewayConf, logger *log.GatewayLog) *CacheGroup {

	g := &CacheGroup{
		RPCGroup: make(map[string]*Cache, 0),
		OWLGroup: make(map[string]*Cache, 0),
	}

	g.rw.Lock()
	defer g.rw.Unlock()

	for n, _ := range config.Rpcmon.BackendMap {
		g.RPCGroup[n] = New(n, config.Rpcmon.AllHistory, logger)
	}

	for n, _ := range config.OWL.BackendMap {
		g.OWLGroup[n] = New(n, config.OWL.AllHistory, logger)
	}

	return g
}

func (g *CacheGroup) Get(name string) *Cache {
	g.rw.RLock()
	defer g.rw.RUnlock()

	if g.RPCGroup == nil || g.OWLGroup == nil {
		return nil
	}

	c, ok := g.RPCGroup[name]
	if !ok {
		c, ok := g.OWLGroup[name]
		if !ok {
			return nil
		}
		return c
	}
	return c
}

func (g *CacheGroup) Close() {

	g.rw.Lock()
	defer g.rw.Unlock()

	for _, n := range g.RPCGroup {
		n.Close()
	}

	g.RPCGroup = nil

	for _, n := range g.OWLGroup {
		n.Close()
	}

	g.OWLGroup = nil

	return
}

func New(name string, allHistory time.Duration, logger *log.GatewayLog) *Cache {
	cache := &Cache{
		Name:          name,
		log:           logger.Named(name),
		checkInterval: 1 * time.Second,
		allHistory:    allHistory,
		quit:          make(chan bool, 0),
	}

	cache.checkExpire()

	return cache
}

func (c *Cache) checkExpire() {

	c.wg.Add(1)

	go func() {
		defer c.wg.Done()

		ticker := time.NewTicker(c.checkInterval)
		defer ticker.Stop()

		for {
			select {
			case <-c.quit:
				return
			case <-ticker.C:
				c.doExpire()
			}
		}
	}()
}

func (c *Cache) doExpire() {

	c.entries.Range(func(k, v interface{}) bool {

		defer func() {
			if p := recover(); p != nil {
				c.log.Error("expire cache panic", zap.String("type", c.Name))
			}
		}()

		service := k.(string)
		entry, _ := v.(*Items)

		entry.mu.Lock()
		defer entry.mu.Unlock()

		if c.Name == "nginx" || c.Name == "exception" {
			newMap := make(map[string]time.Time, 0)
			now := time.Now()
			for id, ts := range entry.uniqMap {
				if ts.Before(now) == false {
					newMap[id] = ts
				} else {
					c.log.Debug("logid expired", zap.String("type", c.Name), zap.String("service", service), zap.String("logid", id), zap.Time("now", now), zap.Time("deadline", ts))
				}
			}
			entry.uniqMap = newMap
		}

		if len(entry.array) == 0 {
			return true
		}

		cnt := 0
		for _, item := range entry.array {
			if item.isExpired() == false {
				break
			}
			cnt += 1
		}

		if cnt == 0 {
			return true
		}

		//item := entry.array[cnt-1]

		if len(entry.array) == cnt {
			cnt -= 1
		}

		if cnt == 0 {
			return true
		}

		//c.log.Debug("expired count", zap.String("type", c.Name), zap.String("service", service), zap.Int("count", cnt), zap.Int("total", len(entry.array)), zap.Duration("expire", item.expire), zap.Time("deadline", item.Create.Add(item.expire)))

		entry.array = entry.array[cnt:]

		atomic.AddUint64(&c.count, ^uint64(cnt-1))

		return true
	})

	return
}

func (c *Cache) load(key string) *Items {

	m, _ := time.ParseDuration("-24h")
	new := &Items{
		array:       make([]*Item, 0),
		waiting:     make(map[string]*CacheRequest, 0),
		lastHistory: time.Now().Add(30 * m),
		lastOne:     time.Now().Add(30 * m),
		uniqMap:     make(map[string]time.Time, 0),
	}

	v, _ := c.entries.LoadOrStore(key, new)

	entry := v.(*Items)

	return entry
}

func (c *Cache) SetAll(allNodes []Node, create time.Time, expire time.Duration, interval time.Duration) error {
	err := c.Set("_all_", allNodes, create, expire, interval)
	if err != nil {
		return err
	}

	for _, v := range allNodes {
		k, ok := v["Service"]
		if !ok {
			return fmt.Errorf("invalid data format:%#v", v)
		}
		val := []Node{v}
		err = c.Set(k.(string), val, create, expire, interval)
		if err != nil {
			return err
		}
	}

	return nil
}

func (c *Cache) Set(key string, val []Node, create time.Time, expire time.Duration, interval time.Duration) error {

	defer func() {
		if p := recover(); p != nil {
			c.log.Error("set cache panic", zap.String("type", c.Name))
		}
	}()

	select {
	case <-c.quit:
		return errors.New("storage is closed")
	default:
	}

	now := time.Now()

	//过期数据
	if expire > 0 && now.Sub(create) > expire {
		c.log.Warn("cant cache, data expired ", zap.String("key", key), zap.Any("value", val), zap.Duration("expire", expire), zap.Time("create", create), zap.Time("now", now))
		return nil
	}

	item := &Item{
		Data:   val,
		expire: expire,
		Create: create,
	}

	entry := c.load(key)

	entry.mu.Lock()
	defer entry.mu.Unlock()

	if c.Name == "exception" || c.Name == "nginx" {
		logID := val[0]["LogID"].(string)
		if _, ok := entry.uniqMap[logID]; ok {
			c.log.Warn("cant cache, log id exists", zap.String("type", c.Name), zap.String("key", key), zap.Any("value", val))
			return nil
		}
		entry.uniqMap[logID] = create.Add(expire)
	}

	//比当前早的数据
	if create.Unix() < entry.lastOne.Unix() {
		c.log.Warn("cant cache, data before current", zap.String("key", key), zap.Time("create", create), zap.Time("lastOne", entry.lastOne), zap.Any("value", val))
		return nil
	}
	entry.lastOne = create

	if create.Sub(entry.lastHistory) >= interval {
		atomic.AddUint64(&c.count, 1)
		entry.array = append(entry.array, item)
		sort.Sort(ItemGroup(entry.array))
		entry.lastHistory = create
		//c.log.Debug("set history item", zap.String("key", key), zap.Time("create", create), zap.Time("lastOne", entry.lastOne), zap.Time("lastHistory", entry.lastHistory), zap.Duration("expire", expire), zap.Duration("interval", interval))
	} else {
		c.log.Warn("cant cache, must greater than interval", zap.String("key", key), zap.Time("create", create), zap.Time("lastHistory", entry.lastHistory), zap.Duration("interval", interval), zap.Any("value", val))
	}

	for remote, req := range entry.waiting {
		select {
		case req.Send <- item:
			c.log.Debug("send backend message done", zap.String("remote", remote), zap.String("type", c.Name))
		default:
			c.log.Debug("send backend message full", zap.String("remote", remote), zap.String("type", c.Name))
		}
	}

	return nil
}

func (c *Cache) GetAll(remote string) (*CacheResponse, error) {
	return c.Get("_all_", remote)
}

func (c *Cache) Get(key string, remote string) (*CacheResponse, error) {

	defer func() {
		if p := recover(); p != nil {
			c.log.Error("get cache panic", zap.String("type", c.Name))
		}
	}()

	select {
	case <-c.quit:
		return nil, errors.New("storage is closed")
	default:
	}

	center := make(chan *Item, 16)

	resp := &CacheResponse{
		Remote:  remote,
		Receive: center,
		History: make([]*Item, 0),
	}

	req := &CacheRequest{
		Send: center,
	}

	entry := c.load(key)

	entry.mu.Lock()
	defer entry.mu.Unlock()

	for _, item := range entry.array {
		if key == "_all_" {
			if c.allHistory == 0 {
				resp.History = append(resp.History, item)
				c.log.Debug("get history", zap.String("key", key), zap.Bool("isExpired", item.isExpired()), zap.Time("create", item.Create), zap.Time("now", time.Now()), zap.Duration("allHistory", c.allHistory))
			} else if c.allHistory > 0 && time.Now().Sub(item.Create) <= c.allHistory {
				resp.History = append(resp.History, item)
				c.log.Debug("get history", zap.String("key", key), zap.Bool("isExpired", item.isExpired()), zap.Time("create", item.Create), zap.Time("now", time.Now()), zap.Duration("allHistory", c.allHistory))
			}
		} else {
			resp.History = append(resp.History, item)
			c.log.Debug("get history", zap.String("key", key), zap.Bool("isExpired", item.isExpired()), zap.Time("create", item.Create), zap.Time("now", time.Now()), zap.Duration("allHistory", c.allHistory))
		}
	}

	entry.waiting[resp.Remote] = req

	return resp, nil
}

func (c *Cache) Cancel(key string, remote string) error {

	defer func() {
		if p := recover(); p != nil {
			c.log.Error("cancel cache panic", zap.String("type", c.Name))
		}
	}()

	select {
	case <-c.quit:
		return errors.New("storage is closed")
	default:
	}

	v, ok := c.entries.Load(key)
	if !ok {
		return errors.New("key does not exist")
	}

	c.log.Debug("unsubscribe cache", zap.String("key", key), zap.String("remote", remote))

	entry := v.(*Items)

	entry.mu.Lock()
	defer func() {
		entry.mu.Unlock()
	}()

	if req, ok := entry.waiting[remote]; ok {
		close(req.Send)
	}
	delete(entry.waiting, remote)

	return nil
}

func (c *Cache) Close() {
	close(c.quit)
	c.wg.Wait()
	c.entries.Range(func(k, v interface{}) bool {
		key := k.(string)
		entry := v.(*Items)

		entry.mu.Lock()
		defer entry.mu.Unlock()

		for _, req := range entry.waiting {
			close(req.Send)
		}
		entry.waiting = nil

		entry.array = nil

		c.entries.Delete(k)

		c.log.Debug("delete cache", zap.String("key", key))

		return true
	})
}
